package mpt;

import java.io.IOException;
import java.io.UnsupportedEncodingException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

public class bg {

	private static Configuration config = null;
	/**
	 * 初始化配置
	 */
	static {
		config = HBaseConfiguration.create();
	}

	public static void main(String[] args) {

		try {
			String tableName = "1crawldb_webpage";
			String[] familys = { "ol", "score" };
			createTable(tableName, familys);
			Job job = Job.getInstance(config, "ExampleSummary");
			job.setJarByClass(bg.class); // class that contains mapper
											// and
											// reducer

			Scan scan = new Scan();
			scan.setCaching(500); // 1 is the default in Scan, which will be
									// bad
									// for MapReduce jobs
			scan.setCacheBlocks(false); // don't set to true for MR jobs
			scan.setBatch(1000);
			// set other scan attrs
			// scan.addColumn(family, qualifier);
			TableMapReduceUtil.initTableMapperJob("crawldb_webpage", // input
																		// table
					scan, // Scan instance to control CF and attribute
							// selection
					MyMapper.class, // mapper class
					Text.class, // mapper output key
					Text.class, // mapper output value
					job);
			TableMapReduceUtil.initTableReducerJob(tableName, // output
					// table
					MyTableReducer.class, // reducer class
					job);
			job.setNumReduceTasks(1); // at least one, adjust as required

			job.waitForCompletion(true);

			System.out.println("==========END========");
			// viewTable(tableName);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void viewTable(String tableName) throws IOException {
		HTable htable = new HTable(config, tableName);
		Scan scan = new Scan();
		ResultScanner scanner = htable.getScanner(scan);
		for (Result r : scanner) {
			System.out.println(new String(r.getRow()));
			System.out.println(new String(r.getValue(Bytes.toBytes("ol"),
					Bytes.toBytes(""))));
		}
		htable.close();
	}

	public static String hbase2url(String url) {

		String hostZA = url.substring(0, url.indexOf(":"));
		url = url.substring(url.indexOf(":") + 1);
		int end = url.lastIndexOf("/") == url.length() - 1 ? url.length() - 1
				: url.length();
		String path = url.substring(url.indexOf("/"), end);
		url = url.substring(0, url.indexOf("/")) + "://";
		String[] hosts = hostZA.split("\\.");
		for (int i = hosts.length - 1; i > 0; i--) {
			url += (hosts[i] + ".");
		}
		return url += (hosts[0] + path);
	}

	public static void createTable(String tableName, String familys[])
			throws Exception {

		HBaseAdmin admin = new HBaseAdmin(config); // 客户端管理工具类
		if (admin.tableExists(tableName)) {
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
			System.out.println("delete table " + tableName + " ok.");
		}
		HTableDescriptor tableDesc = new HTableDescriptor(
				TableName.valueOf(tableName));
		for (int j = 0; j < familys.length; j++) {
			tableDesc.addFamily(new HColumnDescriptor(familys[j]));
		}
		admin.createTable(tableDesc);
		admin.close();
		System.out.println("create table " + tableName + " ok.");
	}

	public static String getFamily(Result value, String family) {
		String bookNameText = new String();
		byte[] cnts = value.getValue(Bytes.toBytes("f"), Bytes.toBytes("cnt"));
		if (cnts == null || cnts.length == 0)
			return bookNameText;
		String cnt = null;
		try {
			cnt = new String(cnts, "GBK");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		if (cnt == null || cnt.equals(""))
			return bookNameText;
		Document doc = Jsoup.parse(cnt);
		Elements metas = doc.select("meta[name=description]");
		if (metas.size() == 0)
			return bookNameText;
		String metaText = metas.attr("content");
		if (metaText == null || metaText.equals(""))
			return bookNameText;
		int s = metaText.indexOf("《");
		int d = metaText.indexOf("》");
		if (s != -1 && d != -1)
			bookNameText = metaText.substring(s + 1, d);

		Elements auth = doc.select("td.wordheight");

		String authText = "";
		for (Element a : auth) {
			if (a.text().equals("出 版 社：")) {
				System.out.print(a.text());
				System.out.println(a.nextElementSibling().child(0).text());
			} else {
				System.out.print(a.text());
				System.out.println(a.nextElementSibling().text());
			}
		}
		if (authText == null || authText.equals(""))
			return bookNameText;
		s = metaText.indexOf("《");
		d = metaText.indexOf("》");
		if (s != -1 && d != -1)
			bookNameText = metaText.substring(s + 1, d);
		return bookNameText;
	}

	public static void jsoup() {

	}

	public static class MyMapper extends TableMapper<Text, Text> {

		public void map(ImmutableBytesWritable row, Result value,
				Context context) throws IOException, InterruptedException {
			String url = Bytes.toString(row.get());
			url = hbase2url(url);
			String bookNameText = getFamily(value, "f");
			if (bookNameText != null && !bookNameText.equals("")) {
				context.write(new Text(bookNameText), new Text(url));
			}
		}
	}

	public static class MyTableReducer extends
			TableReducer<Text, Text, ImmutableBytesWritable> {
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Put put = new Put(Bytes.toBytes(key.toString()));
			for (Text val : values) {
				put.add(Bytes.toBytes("ol"), Bytes.toBytes(""),
						Bytes.toBytes(val.toString()));
			}
			context.write(null, put);
		}
	}
}